package com.fs.config;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author ：hyq
 * @date ：Created in 2020/8/14 10:21
 * @description：
 * @modified By：
 */
public class ProducerinterceptorPrefix implements
        ProducerInterceptor<String, String> {

    private volatile long sendSuccess = 0;
    private volatile long sendFailure = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {

        String modifiedValue = "alarm-" + producerRecord.value();
        return new ProducerRecord<>(producerRecord.topic (),
                producerRecord.partition() , producerRecord.timestamp(),
                producerRecord.key() , modifiedValue, producerRecord.headers()) ;
    }
    /**
     * @Description: 计数统计成功失败的次数
     * @Author: huyunqiang
     * @Date: 2020/8/14 11:16
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            sendSuccess++;
        } else {
            sendFailure ++;
        }
    }
    @Override
    public void close() {
        double successRatio = (double)sendSuccess / (sendFailure + sendSuccess);
        System.out.println ("[INFO]发送成功率＝" + String.format ("%f", successRatio * 100) + "%");
    }
    @Override
    public void configure(Map<String, ?> map) {

    }
}
